package com.young.sync.listener;

import com.young.canal.serializer.EventHeader;
import com.young.canal.serializer.EventRecord;
import com.young.springboot.kafka.RecordFilter;
import com.young.springboot.kafka.RegexRecordFilter;
import com.young.sync.factory.MessageListenerFactory;
import com.young.sync.service.entity.TopicEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 5:26 PM
 */
@Slf4j
public class MessageListener implements BatchAcknowledgingMessageListener<String, EventRecord> {

    private RecordFilter<EventRecord> filter;

    private String bizName;

    public MessageListener(String bizName) {
        this.bizName = bizName;
    }


    public RecordFilter<EventRecord> buildFilter() {
        String filter = null;

        TopicEntity topic = MessageListenerFactory.getMessageListenerByTopic(bizName);

        if (Objects.nonNull(topic)) {
            filter = topic.getFilterRegex();
        }

        log.info("topic:{},filter:{}", bizName,filter);

        return new RegexRecordFilter<EventRecord>(filter) {
            @Override
            public String resolveSource(EventRecord eventRecord) {
                EventHeader header = eventRecord.getHeader();

                return header.getSchemaName() + "." + header.getTableName();
            }
        };
    }

    @Override
    public void onMessage(List<ConsumerRecord<String, EventRecord>> list, Acknowledgment acknowledgment) {


        List<EventRecord> records = new ArrayList<>();

        filter = buildFilter();

        for (ConsumerRecord<String, EventRecord> consumerRecord : list) {
            if (!filter.filter(consumerRecord.value())) {
                records.add(consumerRecord.value());
            }
        }


        if (!records.isEmpty()) {
            //todo 在这里对records业务处理
        }


    }
}
